use serde::{Deserialize, Serialize};
use std::cell::Cell;
use std::sync::Arc;
use tokio::sync::Mutex;

use async_trait::async_trait;
use flyway::{
    ChangelogFile, MigrationExecutor, MigrationState, MigrationStateManager, MigrationStatus,
    MigrationsError,
};
use rbatis::executor::RBatisTxExecutor;
use rbatis::rbdc::datetime::DateTime;
use rbatis::RBatis;
use rbs::value;

/// Default table name for the migration state management table
pub const DEFAULT_MIGRATIONS_TABLE: &str = "flyway_migrations";

#[derive(Clone, Debug, Serialize, Deserialize)]
struct MigrationInfo {
    ts: DateTime,
    version: u32,
    name: Option<String>,
    checksum: Option<String>,
    status: Option<String>,
}
/// Available driver types supported by Rbatis
pub enum RbatisDbDriverType {
    MySql,
    Pg,
    Sqlite,
    MsSql,
    TDengine,
    Dameng,
    Other(String),
}
/// 不同数据库的建表语句
fn create_table_sql(db_type: RbatisDbDriverType, migrations_table_name: String) -> String {
    match db_type {
        RbatisDbDriverType::MySql => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(64) null,
                status VARCHAR(16)
            );"#,
                migrations_table_name
            )
        }
        RbatisDbDriverType::Pg => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(64) null,
                status VARCHAR(16)
            );"#,
                migrations_table_name
            )
        }
        RbatisDbDriverType::Sqlite => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(64) null,
                status VARCHAR(16)
            );"#,
                migrations_table_name
            )
        }
        RbatisDbDriverType::MsSql => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(64) null,
                status VARCHAR(16)
            );"#,
                migrations_table_name
            )
        }
        RbatisDbDriverType::TDengine => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (`ts` TIMESTAMP, `version` int,`name` nchar(255) , `checksum` nchar(64), `status` nchar(255))
                  "#,
                migrations_table_name
            )
        }
        RbatisDbDriverType::Other(_) => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(64) null,
                status VARCHAR(16)
            );"#,
                migrations_table_name
            )
        }
        RbatisDbDriverType::Dameng => {
            format!(
                r#"CREATE TABLE IF NOT EXISTS {} (
                version INTEGER PRIMARY KEY,
                ts       varchar(255) null,
                name     varchar(255) null,
                checksum   varchar(64) null,
                status VARCHAR(16)
            );"#,
                migrations_table_name
            )
        }
    }
}
/// 不同数据库的update
fn update_sql(
    db_type: RbatisDbDriverType,
    migrations_table_name: String,
    status: String,
    version: u64,
) -> String {
    match db_type {
        RbatisDbDriverType::MySql => {
            format!(
                r#"UPDATE {} SET status='{}' where version={};"#,
                migrations_table_name.as_str(),
                status.as_str(),
                version
            )
        }
        RbatisDbDriverType::Pg => {
            format!(
                r#"UPDATE {} SET status='{}' where version={};"#,
                migrations_table_name.as_str(),
                status.as_str(),
                version
            )
        }
        RbatisDbDriverType::Sqlite => {
            unimplemented!()
        }
        RbatisDbDriverType::MsSql => {
            unimplemented!()
        }
        RbatisDbDriverType::TDengine => {
            unimplemented!()
        }
        RbatisDbDriverType::Dameng => {
            format!(
                r#"UPDATE {} SET status='{}' where version={};"#,
                migrations_table_name.as_str(),
                status.as_str(),
                version
            )
        }
        RbatisDbDriverType::Other(_) => {
            unimplemented!()
        }
    }
}
/// 不同数据库的insert
fn insert_sql(
    db_type: RbatisDbDriverType,
    migrations_table_name: String,
    status: String,
) -> String {
    match db_type {
        RbatisDbDriverType::MySql => {
            format!(
                r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                migrations_table_name.as_str(),
                status.as_str()
            )
        }
        RbatisDbDriverType::Pg => {
            format!(
                r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                migrations_table_name.as_str(),
                status.as_str()
            )
        }
        RbatisDbDriverType::Sqlite => {
            unimplemented!()
        }
        RbatisDbDriverType::MsSql => {
            unimplemented!()
        }
        RbatisDbDriverType::TDengine => {
            unimplemented!()
        }
        RbatisDbDriverType::Other(_) => {
            unimplemented!()
        }
        RbatisDbDriverType::Dameng => {
            format!(
                r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, '{}');"#,
                migrations_table_name.as_str(),
                status.as_str()
            )
        }
    }
}

/// Rbatis implementation of `MigrationStateManager` and `MigrationExecutor`
pub struct RbatisMigrationDriver {
    db: Arc<RBatis>,
    migrations_table_name: String,
    tx: Mutex<Cell<Option<RBatisTxExecutor>>>,
}

impl RbatisMigrationDriver {
    /// Create a new driver
    ///
    ///  * `db`: The `Rbatis` instance for accessing the database
    ///  * `migrations_table_name`: The optional name of the table the migration state information
    ///    should be stored in. If `None`, the `DEFAULT_MIGRATIONS_TABLE` will be used.
    pub fn new(db: Arc<RBatis>, migrations_table_name: Option<&str>) -> RbatisMigrationDriver {
        return RbatisMigrationDriver {
            db: db.clone(),
            migrations_table_name: migrations_table_name
                .map(|v| v.to_string())
                .or(Some(DEFAULT_MIGRATIONS_TABLE.to_string()))
                .unwrap(),
            tx: Mutex::new(Cell::new(None)),
        };
    }

    /// The the driver type of the `Rbatis` instance
    ///
    /// This method will get the driver type from `Rbatis` (which is a string) and convert it into
    /// an `RbatisDbDriverType`. `Other(String)` will be used for any database drivers not directly
    /// known to `flyway-rbatis`.
    pub fn driver_type(&self) -> rbatis::Result<RbatisDbDriverType> {
        let db = self.db.clone();
        let driver_type_name = db.driver_type()?;
        let result = match driver_type_name {
            "mssql" => RbatisDbDriverType::MsSql,
            "mysql" => RbatisDbDriverType::MySql,
            "postgres" => RbatisDbDriverType::Pg,
            "sqlite" => RbatisDbDriverType::Sqlite,
            "Taos" => RbatisDbDriverType::TDengine,
            "Dameng" => RbatisDbDriverType::Dameng,
            "dm" => RbatisDbDriverType::Dameng,
            "dameng" => RbatisDbDriverType::Dameng,
            _ => RbatisDbDriverType::Other(driver_type_name.to_string()),
        };
        return Ok(result);
    }

    async fn object_exist(&self, sql: &str, args: Vec<rbs::Value>) -> flyway::Result<bool> {
        let db = self.db.clone();
        if let Ok(CountDto { count }) = db.query_decode::<CountDto>(sql, args).await {
            if count > 0 {
                return Ok(true);
            }
        }
        Ok(false)
    }
}

/// Implementation of the `MigrationStateManager`
#[async_trait]
impl MigrationStateManager for RbatisMigrationDriver {
    async fn prepare(&self) -> flyway::Result<()> {
        log::debug!("Preparing Migrations Table ...");
        let db = self.db.clone();
        let statement = create_table_sql(
            self.driver_type().unwrap(),
            self.migrations_table_name.clone(),
        );
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;

        log::debug!("Preparation Statement: {}", statement.as_str());
        let _result = db
            .exec(statement.as_str(), vec![])
            .await
            .or_else(|err| Err(MigrationsError::migration_setup_failed(Some(err.into()))))?;
        log::debug!("Preparing Migrations Table ... done");
        return Ok(());
    }

    async fn lowest_version(&self) -> flyway::Result<Option<MigrationState>> {
        log::debug!("Retrieving lowest version ... ");
        let db = self.db.clone();
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;
        let version: Option<u64> = db
            .query_decode(
                format!(
                    "SELECT MIN(version) FROM {} WHERE status='deployed';",
                    self.migrations_table_name.as_str()
                )
                .as_str(),
                vec![],
            )
            .await
            .or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?;

        log::debug!("Retrieving lowest version ... {:?}", &version);
        return Ok(version.and_then(|version| {
            Some(MigrationState {
                version,
                status: MigrationStatus::Deployed,
            })
        }));
    }

    async fn highest_version(&self) -> flyway::Result<Option<MigrationState>> {
        log::debug!("Retrieving highest version ... ");
        let db = self.db.clone();
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;
        let version: Option<u64> = db
            .query_decode(
                format!(
                    "SELECT MAX(version) FROM {} WHERE status='deployed';",
                    self.migrations_table_name.as_str()
                )
                .as_str(),
                vec![],
            )
            .await
            .or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?;

        log::debug!("Retrieving highest version ... {:?}", &version);
        return Ok(version.and_then(|version| {
            Some(MigrationState {
                version,
                status: MigrationStatus::Deployed,
            })
        }));
    }

    async fn list_versions(&self) -> flyway::Result<Vec<MigrationState>> {
        log::debug!("Listing versions ... ");
        let db = self.db.clone();
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;
        let versions: Vec<u64> = db
            .query_decode(
                format!(
                    "SELECT version FROM {} WHERE status='deployed' ORDER BY version asc;",
                    self.migrations_table_name.as_str()
                )
                .as_str(),
                vec![],
            )
            .await
            .or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?;

        let versions: Vec<MigrationState> = versions
            .iter()
            .map(|version| MigrationState {
                version: *version,
                status: MigrationStatus::Deployed,
            })
            .collect();

        log::debug!("Listing versions ... {:?}", &versions);
        return Ok(versions);
    }

    async fn begin_version(&self, changelog_file: &ChangelogFile) -> flyway::Result<()> {
        log::debug!("Beginning version ... {}", changelog_file.version);
        let db = self.db.clone();
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;

        match self.driver_type() {
            Ok(db_type) => {
                match db_type {
                    RbatisDbDriverType::TDengine => {
                        let mut ts: i64 = DateTime::utc().unix_timestamp_millis()
                            + changelog_file.version() as i64;
                        let ts_select = format!(
                            r#"select ts,version from {} where status='in_progress' and version=? limit 1;"#,
                            self.migrations_table_name.as_str()
                        );
                        match db
                            .query_decode::<Vec<MigrationInfo>>(
                                ts_select.as_str(),
                                vec![value!(changelog_file.version.clone())],
                            )
                            .await
                        {
                            Ok(result) => {
                                // println!("{:?}",result);
                                if result.first().is_some() {
                                    let time = result
                                        .first()
                                        .unwrap()
                                        .ts
                                        .clone()
                                        .set_offset(-16 * 60 * 60);
                                    ts = time.unix_timestamp_millis();
                                }
                            }
                            Err(e) => {
                                log::error!("数据异常:{}", e.to_string())
                            }
                        };
                        let insert_statement = format!(
                            r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
                            self.migrations_table_name.as_str()
                        );
                        log::debug!("Insert statement: {}", insert_statement.as_str());
                        let _insert_result = db
                            .exec(
                                insert_statement.as_str(),
                                vec![
                                    value!(ts),
                                    value!(changelog_file.version.clone()),
                                    value!(changelog_file.name.clone()),
                                    value!(changelog_file.checksum.clone()),
                                ],
                            )
                            .await
                            .or_else(|err| {
                                Err(MigrationsError::migration_versioning_failed(Some(
                                    err.into(),
                                )))
                            })?;
                        return Ok(());
                    }
                    _ => {}
                }
            }
            Err(_) => {}
        }

        // let update_statement = format!(r#"UPDATE {} SET status='in_progress' where version={};"#,
        //                                self.migrations_table_name.as_str(), changelog_file.version);
        let update_statement = update_sql(
            self.driver_type().unwrap(),
            self.migrations_table_name.clone(),
            "in_progress".to_string(),
            changelog_file.version.clone(),
        );

        log::debug!("Update statement: {}", update_statement.as_str());
        let update_result = db
            .exec(update_statement.as_str(), vec![])
            .await
            .or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?;

        if update_result.rows_affected < 1 {
            let ts: i64 = DateTime::utc().unix_timestamp_millis() + changelog_file.version as i64;

            // let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
            //                                self.migrations_table_name.as_str());
            let insert_statement = insert_sql(
                self.driver_type().unwrap(),
                self.migrations_table_name.clone(),
                "in_progress".to_string(),
            );
            log::debug!("Insert statement: {}", insert_statement.as_str());
            let _insert_result = db
                .exec(
                    insert_statement.as_str(),
                    vec![
                        value!(ts),
                        value!(changelog_file.version.clone()),
                        value!(changelog_file.name.clone()),
                        value!(changelog_file.checksum.clone()),
                    ],
                )
                .await
                .or_else(|err| {
                    Err(MigrationsError::migration_versioning_failed(Some(
                        err.into(),
                    )))
                })?;
        }

        return Ok(());
    }

    async fn finish_version(&self, changelog_file: &ChangelogFile) -> flyway::Result<()> {
        log::debug!("Finishing version ... {}", changelog_file.version);
        log::info!("{} execution completed.", changelog_file.name);
        let db = self.db.clone();
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;

        match self.driver_type() {
            Ok(db_type) => match db_type {
                RbatisDbDriverType::TDengine => {
                    let mut ts: i64 =
                        DateTime::utc().unix_timestamp_millis() + changelog_file.version as i64;
                    let ts_select = format!(
                        r#"select ts,version from {} where status='in_progress' and version=? limit 1;"#,
                        self.migrations_table_name.as_str()
                    );
                    match db
                        .query_decode::<Vec<MigrationInfo>>(
                            ts_select.as_str(),
                            vec![value!(changelog_file.version.clone())],
                        )
                        .await
                    {
                        Ok(result) => {
                            if result.first().is_some() {
                                let time =
                                    result.first().unwrap().ts.clone().set_offset(-16 * 60 * 60);
                                ts = time.unix_timestamp_millis();
                            }
                        }
                        Err(e) => {
                            log::error!("数据异常:{}", e.to_string())
                        }
                    };

                    let insert_statement = format!(
                        r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'deployed');"#,
                        self.migrations_table_name.as_str()
                    );
                    log::debug!("Insert statement: {}", insert_statement.as_str());
                    let _insert_result = db
                        .exec(
                            insert_statement.as_str(),
                            vec![
                                value!(ts),
                                value!(changelog_file.version.clone()),
                                value!(changelog_file.name.clone()),
                                value!(changelog_file.checksum.clone()),
                            ],
                        )
                        .await
                        .or_else(|err| {
                            Err(MigrationsError::migration_versioning_failed(Some(
                                err.into(),
                            )))
                        })?;
                    return Ok(());
                }
                _ => {}
            },
            Err(_) => {}
        }

        // let update_statement = format!(r#"UPDATE {} SET status='deployed' where version={};"#,
        //                                self.migrations_table_name.as_str(), changelog_file.version);
        let update_statement = update_sql(
            self.driver_type().unwrap(),
            self.migrations_table_name.clone(),
            "deployed".to_string(),
            changelog_file.version.clone(),
        );

        log::debug!("Update statement: {}", update_statement.as_str());
        let update_result = db
            .exec(update_statement.as_str(), vec![])
            .await
            .or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?;

        if update_result.rows_affected < 1 {
            let ts: i64 = DateTime::utc().unix_timestamp_millis() + changelog_file.version as i64;

            // let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
            //                                self.migrations_table_name.as_str());
            let insert_statement = insert_sql(
                self.driver_type().unwrap(),
                self.migrations_table_name.clone(),
                "in_progress".to_string(),
            );

            log::debug!("Insert statement: {}", insert_statement.as_str());
            let _insert_result = db
                .exec(
                    insert_statement.as_str(),
                    vec![
                        value!(ts),
                        value!(changelog_file.version.clone()),
                        value!(changelog_file.name.clone()),
                        value!(changelog_file.checksum.clone()),
                    ],
                )
                .await
                .or_else(|err| {
                    Err(MigrationsError::migration_versioning_failed(Some(
                        err.into(),
                    )))
                })?;
        }

        return Ok(());
    }

    async fn skip_version(&self, changelog_file: &ChangelogFile) -> flyway::Result<()> {
        log::debug!("Skip version ... {}", changelog_file.version);
        let db = self.db.clone();
        let db = db.acquire().await.or_else(|err| {
            Err(MigrationsError::migration_database_failed(
                None,
                Some(err.into()),
            ))
        })?;

        match self.driver_type() {
            Ok(db_type) => match db_type {
                RbatisDbDriverType::TDengine => {
                    let mut ts: i64 =
                        DateTime::utc().unix_timestamp_millis() + changelog_file.version as i64;
                    let ts_select = format!(
                        r#"select ts,version from {} where status='in_progress' and version=? limit 1;"#,
                        self.migrations_table_name.as_str()
                    );
                    match db
                        .query_decode::<Vec<MigrationInfo>>(
                            ts_select.as_str(),
                            vec![value!(changelog_file.version.clone())],
                        )
                        .await
                    {
                        Ok(result) => {
                            if result.first().is_some() {
                                let time =
                                    result.first().unwrap().ts.clone().set_offset(-16 * 60 * 60);
                                ts = time.unix_timestamp_millis();
                            }
                        }
                        Err(e) => {
                            log::error!("数据异常:{}", e.to_string())
                        }
                    };

                    let insert_statement = format!(
                        r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'fail');"#,
                        self.migrations_table_name.as_str()
                    );
                    log::debug!("Insert statement: {}", insert_statement.as_str());
                    let _insert_result = db
                        .exec(
                            insert_statement.as_str(),
                            vec![
                                value!(ts),
                                value!(changelog_file.version.clone()),
                                value!(changelog_file.name.clone()),
                                value!(changelog_file.checksum.clone()),
                            ],
                        )
                        .await
                        .or_else(|err| {
                            Err(MigrationsError::migration_versioning_failed(Some(
                                err.into(),
                            )))
                        })?;
                    return Ok(());
                }
                _ => {}
            },
            Err(_) => {}
        }

        // let update_statement = format!(r#"UPDATE {} SET status='deployed' where version={};"#,
        //                                self.migrations_table_name.as_str(), changelog_file.version);
        let update_statement = update_sql(
            self.driver_type().unwrap(),
            self.migrations_table_name.clone(),
            "fail".to_string(),
            changelog_file.version.clone(),
        );

        log::debug!("Update statement: {}", update_statement.as_str());
        let update_result = db
            .exec(update_statement.as_str(), vec![])
            .await
            .or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?;

        if update_result.rows_affected < 1 {
            let ts: i64 = DateTime::utc().unix_timestamp_millis() + changelog_file.version as i64;

            // let insert_statement = format!(r#"INSERT INTO {}(ts,version,name,checksum, status) VALUES (?,?,?,?, 'in_progress');"#,
            //                                self.migrations_table_name.as_str());
            let insert_statement = insert_sql(
                self.driver_type().unwrap(),
                self.migrations_table_name.clone(),
                "in_progress".to_string(),
            );

            log::debug!("Insert statement: {}", insert_statement.as_str());
            let _insert_result = db
                .exec(
                    insert_statement.as_str(),
                    vec![
                        value!(ts),
                        value!(changelog_file.version.clone()),
                        value!(changelog_file.name.clone()),
                        value!(changelog_file.checksum.clone()),
                    ],
                )
                .await
                .or_else(|err| {
                    Err(MigrationsError::migration_versioning_failed(Some(
                        err.into(),
                    )))
                })?;
        }

        return Ok(());
    }

    async fn merge_seaql_migration(&self, changelog_file: &ChangelogFile) -> flyway::Result<bool> {
        let sql = match self.driver_type() {
            Ok(db_type) => match db_type {
                RbatisDbDriverType::Dameng => {
                     "SELECT COUNT(*) as count  FROM ALL_OBJECTS WHERE OBJECT_NAME = 'seaql_migrations' AND OBJECT_TYPE = 'TABLE'"
                },
                RbatisDbDriverType::MsSql => { "SELECT COUNT(*) as count FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'seaql_migrations'"},
                RbatisDbDriverType::Pg => {
                     "SELECT COUNT(*) as count FROM pg_catalog.pg_tables WHERE schemaname = CURRENT_SCHEMA AND tablename = 'seaql_migrations'"
                },
                _ =>  "SELECT COUNT(*) as count FROM pg_catalog.pg_tables WHERE schemaname = CURRENT_SCHEMA AND tablename = 'seaql_migrations'",
            },
            Err(_) =>  "SELECT COUNT(*) as count FROM pg_catalog.pg_tables WHERE schemaname = CURRENT_SCHEMA AND tablename = 'seaql_migrations'"
        };
        if self.object_exist(sql, vec![]).await? {
            if self
                .object_exist(
                    r#"SELECT COUNT(*) as count FROM seaql_migrations AS sm where version = ? "#,
                    vec![rbs::value!(changelog_file.name.clone())],
                )
                .await?
            {
                println!(
                    "seaql_migrations表中已执行版本: {}, flyway 跳过名称为{}的版本.....",
                    changelog_file.name.clone(),
                    changelog_file.name.clone()
                );
                let timestamp = DateTime::utc().unix_timestamp_millis();
                let sql = r#"INSERT INTO flyway_migrations(ts,version,name,checksum, status) VALUES (?,?,?,?, 'deployed');"#;
                let _ = self
                    .db
                    .clone()
                    .exec(
                        &sql,
                        vec![
                            value!(timestamp),
                            value!(changelog_file.version.clone()),
                            value!(changelog_file.name.clone()),
                            value!(changelog_file.checksum.clone()),
                        ],
                    )
                    .await
                    .or_else(|err| {
                        Err(MigrationsError::migration_versioning_failed(Some(
                            err.into(),
                        )))
                    })?;
                return Ok(true);
            }
        }
        Ok(false)
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct CountDto {
    pub(crate) count: usize,
}

/// Implementation of the `MigrationExecutor`
#[async_trait]
impl MigrationExecutor for RbatisMigrationDriver {
    async fn begin_transaction(&self) -> flyway::Result<()> {
        log::debug!("Beginning transaction ...");
        {
            let mut tx_guard = self.tx.lock().await;
            if tx_guard.get_mut().is_some() {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        }

        let tx = {
            let db = self.db.clone();
            db.acquire_begin().await.or_else(|err| {
                Err(MigrationsError::migration_versioning_failed(Some(
                    err.into(),
                )))
            })?
        };

        let tx_guard = self.tx.lock().await;
        tx_guard.set(Some(tx));
        return Ok(());
    }

    async fn execute_changelog_file(
        &self,
        changelog_file: &flyway::ChangelogFile,
    ) -> flyway::Result<()> {
        log::debug!("Executing changelog file ... {:?}", &changelog_file);
        let mut tx_guard = self.tx.lock().await;
        let tx = tx_guard.get_mut().as_mut();
        match tx {
            Some(tx) => {
                for statement in changelog_file.to_sql() {
                    // log::debug!("Executing statement: {}", statement.statement.as_str());
                    log::debug!("Executing sql: {}", statement);
                    tx.exec(statement.as_str(), vec![]).await.or_else(|err| {
                        Err(MigrationsError::migration_versioning_failed(Some(
                            err.into(),
                        )))
                    })?;
                }
            }
            None => {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        };
        return Ok(());
    }

    async fn commit_transaction(&self) -> flyway::Result<()> {
        log::debug!("Committing transaction ...");
        let mut tx = {
            let tx_guard = self.tx.lock().await;
            tx_guard.replace(None)
        };

        match tx.as_mut() {
            Some(tx) => {
                return tx.commit().await.map(|_| ()).or_else(|err| {
                    Err(MigrationsError::migration_database_failed(
                        None,
                        Some(err.into()),
                    ))
                });
            }
            None => {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        }
    }

    async fn rollback_transaction(&self) -> flyway::Result<()> {
        log::debug!("Rolling back transaction ...");
        let mut tx = {
            let tx_guard = self.tx.lock().await;
            tx_guard.replace(None)
        };

        match tx.as_mut() {
            Some(tx) => {
                return tx.rollback().await.map(|_| ()).or_else(|err| {
                    Err(MigrationsError::migration_database_failed(
                        None,
                        Some(err.into()),
                    ))
                });
            }
            None => {
                return Err(MigrationsError::migration_database_failed(None, None));
            }
        }
    }
}
